package com.lhkj.ct.framework.log.buffer;

import com.google.common.collect.Lists;
import com.lhkj.ct.base.buffer.Buffer;
import com.lhkj.ct.base.buffer.DataProcessor;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.*;

/**
 * @Description: 缓存统计器
 * @Author:
 * @Date: 2020/6/6 13:15
 */

public class DefaultBuffer implements Buffer<String> {

    private static final Logger logger = LoggerFactory.getLogger(DefaultBuffer.class);
    /**
     * 信号
     */
    private static final String                   SIGNAL       = "signal";
    /**
     * 容量
     */
    private              int                      size;
    /**
     * 处理器
     */
    private DataProcessor processor;
    /**
     * 数据队列
     */
    private              BlockingQueue<String>    queue;
    /**
     * 缓冲区
     */
    private              List<String>             dataList;
    /**
     * 定时线程
     */
    private final ScheduledExecutorService flushExecutor;

    /**
     * 功能描述: 构造函数
     *
     * @param size 缓冲大小
     * @param frequency 频率
     * @param processor 处理器
     */
    public DefaultBuffer(int size, long frequency, DataProcessor processor) {
        this.size = size;
        this.queue = new LinkedBlockingQueue<>();
        this.dataList = Lists.newArrayList();
        this.processor = processor;
        //java定时任务，开一个线程
        this.flushExecutor = Executors.newScheduledThreadPool(1);
        //每多少频率，做什么命令
        this.flushExecutor.scheduleAtFixedRate(() -> queue.offer(SIGNAL), frequency, frequency, TimeUnit.SECONDS);
        Thread listenerThread = new Thread(new BufferStatisticsMsgListener());
        listenerThread.start();
    }

    @Override
    public void put(String data) {
        queue.offer(data);
    }

    @Override
    public void close() {
        queue.offer(SIGNAL);
        flushExecutor.shutdown();
    }

    /**
     * 功能描述: 缓冲统计消息监听器
     *
     */
    private class BufferStatisticsMsgListener implements Runnable {
        @Override
        public void run() {
            logger.info("缓存统计-监听器启动");
            while (true) {
                try {
                    String data = queue.take();
                    if (!StringUtils.equals(data, SIGNAL)) {
                        dataList.add(data);
                    }
                    if (dataList.size() > size || StringUtils.equals(data, SIGNAL)) {
                        if (dataList.size() == 0) {
                            continue;
                        }
                        List<String> buffer = dataList;
                        dataList = Lists.newArrayList();
                        processor.process(buffer);
                        logger.debug("缓存统计-统计消息消费成功");
                    }
                } catch (Throwable e) {
                    logger.error("缓存统计-监听器发生错误", e);
                }
            }
        }
    }

}
